﻿using System;
using System.Text;
using Newtonsoft.Json;
using RabbitMQ.Client;
using RabbitMQ.Client.Events;
using Microsoft.Extensions.Configuration;
using System.Threading.Tasks;

namespace QQZiFramework.MQ
{
    public class RabbitMQ : IMQDriver
    {
        IConnectionFactory _factory;
        //string hostName, port, userName, password;
        bool AutoAck = true;//自动确认还是手动确认（手动只有处理成功的，才会从队列中删除）
        IConfiguration Configuration;

        //ushort BasicQos;
        //bool Durable;//持久化 
        //bool isInited;

        public RabbitMQ(IConfiguration configuration)
        {
            Configuration = configuration;
            //var configItem = configuration.GetSection("RabbitMQ");
            //hostName = configItem["HostName"];
            //port = configItem["Port"];
            //userName = configItem["UserName"];
            //password = configItem["Password"];
            //var maxPoolSize = configItem["maxPoolSize"];
            //var minPoolSize = configItem["minPoolSize"];
            //if (configItem["AutoAck"] != null)
            //    AutoAck = Convert.ToBoolean(configItem["AutoAck"]);
            //if (string.IsNullOrEmpty(hostName)) throw new ArgumentNullException("请在appsetting.json啊配置RabbitMQ:HostName");
            //if (string.IsNullOrEmpty(port)) throw new ArgumentNullException("请在appsetting.json啊配置RabbitMQ:Port");
            //if (string.IsNullOrEmpty(userName)) throw new ArgumentNullException("请在appsetting.json啊配置RabbitMQ:UserName");
            //if (string.IsNullOrEmpty(password)) throw new ArgumentNullException("请在appsetting.json啊配置RabbitMQ:Password");
            //var queues = configItem.GetSection("Queues");
            //if (queues != null)
            //{

            //    if (queues["BasicQos"] != null) BasicQos = Convert.ToUInt16(queues["BasicQos"]);
            //    if (queues["Durable"] != null) Durable = Convert.ToBoolean(queues["Durable"]);

            //}
            //var connectionString = $"amqp://{userName}:{password}@{hostName}:{port}/";
            //if (!isInited)
            //{
            //    isInited = true;
            //    if(maxPoolSize==null) maxPoolSize = "500";
            //    if(minPoolSize==null) minPoolSize = "5";
            //    RabbitMQConnectionPool.Init(connectionString, Convert.ToInt32(minPoolSize), Convert.ToInt32(maxPoolSize));
            //}
        }

        private MQIModel CreateMQIModel(string queueName)
        {
            return RabbitMQChannelPoolManager.GetChannel(queueName);
            // var section = Configuration.GetSection("RabbitMQ:Queues:" + queueName);
            // var conn = RabbitMQConnectShareManage.GetConnection();
            // var channel = conn.connection.CreateModel();//创建连接会话对象
            // bool durable = (section != null && !string.IsNullOrEmpty(section["Durable"])) ? Convert.ToBoolean(section["Durable"]) : Durable;
            // var bindSection = section.GetSection("Bind");
            // string exchange, routingKey = null;
            // string declareQueueName = null;
            // if (bindSection != null)
            // {
            //     exchange = bindSection["exchange"];
            //     routingKey = bindSection["routingKey"];
            //     if (routingKey == null) routingKey = "#";
            //     if (exchange == null) exchange = "";
            //     if (exchange != queueName) declareQueueName = queueName;
            //     else declareQueueName = "";
            // }
            // else exchange = "";
            // if (exchange.Length == 0)
            // {
            //     declareQueueName = queueName; //没有交换机的，申明channel时候传入queue
            //     routingKey = queueName;
            // }
            // //声明一个队列            
            // channel.QueueDeclare(
            //     queue: declareQueueName,
            //      durable: durable,//是否持久化,true持久化,队列会保存磁盘,服务器重启时可以保证不丢失相关信息。
            //      exclusive: false,//是否排他,true排他的,如果一个队列声明为排他队列,该队列仅对首次声明它的连接可见,并在连接断开时自动删除.
            //      autoDelete: false,//是否自动删除。true是自动删除。自动删除的前提是：致少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时,才会自动删除.
            //      arguments: null //设置队列的一些其它参数
            //   );
            // ushort basicQos = (section != null && !string.IsNullOrEmpty(section["BasicQos"])) ? Convert.ToUInt16(section["BasicQos"]) : BasicQos;
            // if (basicQos > 0) channel.BasicQos(0, basicQos, false);
            // IBasicProperties properties = null;
            // if (durable)
            // {
            //     //消息持久化
            //     properties = channel.CreateBasicProperties();
            //     properties.DeliveryMode = 2;
            // }
            // if (exchange.Length > 0)
            // {
            //     //交换机绑定，如果配置名与交换机名一致，这发交换机，如果不一致，发队列
            //     string type = bindSection["exchangeType"];
            //     if (type == null) type = "topic";
            //     channel.ExchangeDeclare(exchange, type, durable);
            //     if (exchange != queueName)
            //     {
            //         //发队列
            //         channel.QueueBind(queueName, exchange, routingKey);
            //     }
            // }
            // return new MQIModel { Conn=conn, Channel = channel, DurableProperties = properties, Exchange = exchange, RoutingKey = routingKey };
        }

        System.Collections.Generic.Dictionary<string, MQIModel> channels = new System.Collections.Generic.Dictionary<string, MQIModel>();

        // private MQIModel CreateMQIModel(string queueName)
        // {

        //     var section = Configuration.GetSection("RabbitMQ:Queues:" + queueName);

        //     var conn = connectionPool.GetConnection();
        //     var channel = conn.CreateModel();//创建连接会话对象
        //     bool durable = (section != null && !string.IsNullOrEmpty(section["Durable"])) ? Convert.ToBoolean(section["Durable"]) : Durable;
        //     var bindSection = section.GetSection("Bind");
        //     string exchange, routingKey = null;
        //     string declareQueueName = null;
        //     if (bindSection != null)
        //     {
        //         exchange = bindSection["exchange"];
        //         routingKey = bindSection["routingKey"];
        //         if (routingKey == null) routingKey = "#";
        //         if (exchange == null) exchange = "";
        //         if (exchange != queueName) declareQueueName = queueName;
        //         else declareQueueName = "";
        //     }
        //     else exchange = "";
        //     if (exchange.Length == 0)
        //     {
        //         declareQueueName = queueName; //没有交换机的，申明channel时候传入queue
        //         routingKey = queueName;
        //     }
        //     //声明一个队列            
        //     channel.QueueDeclare(
        //         queue: declareQueueName,
        //          durable: durable,//是否持久化,true持久化,队列会保存磁盘,服务器重启时可以保证不丢失相关信息。
        //          exclusive: false,//是否排他,true排他的,如果一个队列声明为排他队列,该队列仅对首次声明它的连接可见,并在连接断开时自动删除.
        //          autoDelete: false,//是否自动删除。true是自动删除。自动删除的前提是：致少有一个消费者连接到这个队列，之后所有与这个队列连接的消费者都断开时,才会自动删除.
        //          arguments: null //设置队列的一些其它参数
        //       );
        //     ushort basicQos = (section != null && !string.IsNullOrEmpty(section["BasicQos"])) ? Convert.ToUInt16(section["BasicQos"]) : BasicQos;
        //     if (basicQos > 0) channel.BasicQos(0, basicQos, false);
        //     IBasicProperties properties = null;
        //     if (durable)
        //     {
        //         //消息持久化
        //         properties = channel.CreateBasicProperties();
        //         properties.DeliveryMode = 2;
        //     }
        //     if (exchange.Length > 0)
        //     {
        //         //交换机绑定，如果配置名与交换机名一致，这发交换机，如果不一致，发队列
        //         string type = bindSection["exchangeType"];
        //         if (type == null) type = "topic";
        //         channel.ExchangeDeclare(exchange, type, durable);
        //         if (exchange != queueName)
        //         {
        //             //发队列
        //             channel.QueueBind(queueName, exchange, routingKey);
        //         }
        //     }
        //     return new MQIModel { Conn = conn, Channel = channel, DurableProperties = properties, Exchange = exchange, RoutingKey = routingKey };

        // }
        public void Publish(string queueName, object obj)
        {            
            using (MQIModel item = CreateMQIModel(queueName))
            {
                byte[] body;
                Type type = obj.GetType();
                if (type.IsValueType || type == typeof(string))
                {
                    string itemJson = obj.ToString();
                    body = Encoding.UTF8.GetBytes(itemJson);
                }
                else if (type == typeof(byte[]))
                {
                    body = (byte[])obj;
                }
                else
                {
                    string itemJson;
                    if (obj is IDBEntity) itemJson = (obj as IDBEntity).ToJson();
                    else itemJson = JsonConvert.SerializeObject(obj);
                    body = Encoding.UTF8.GetBytes(itemJson);
                }
                //发送消息
                item.BasicPublish(body);
            }
        }

        public long GetLength(string queueName)
        {
            using (MQIModel item = CreateMQIModel(queueName))
            {
                QueueDeclareOk result = item.Channel.QueueDeclarePassive(queueName);
                long c = (long)result.MessageCount;
                return c;
            }
            
        }

        public void BindReceived(string queueName, Func<byte[], bool> fun)
        {
            if (fun == null) throw new ArgumentNullException("fun");
            var section = Configuration.GetSection("RabbitMQ:Consumers:" + queueName);
            bool autoAck = (section != null && !string.IsNullOrEmpty(section["AutoAck"])) ? Convert.ToBoolean(section["AutoAck"]) : AutoAck;
            UInt16 size = (section != null && !string.IsNullOrEmpty(section["Size"])) ? Convert.ToUInt16(section["Size"]) : (UInt16)1;

            for (UInt16 i = 0; i < size; i++)
            {
                MQIModel item = CreateMQIModel(queueName);
                //创建消费者对象
                var consumer = new EventingBasicConsumer(item.Channel);
                //如果列队有配置aotuack则使用队列的，否则使用全局的
                consumer.Received += (model, ea) =>
                {
                    // byte[] message = ea.Body.ToArray();//接收到的消息
                    // string msg = Encoding.UTF8.GetString(message);
                    bool result = fun.Invoke(ea.Body.ToArray());
                    //Console.WriteLine("接收到消息为:" + Encoding.UTF8.GetString(message));
                    if (!autoAck && result)
                    {
                        //返回消息确认
                        item.Channel.BasicAck(ea.DeliveryTag, false);
                    }
                };
                //消费者开启监听
                item.Channel.BasicConsume(queueName, autoAck, consumer);
            }
        }

        public void BindReceivedAsync(string queueName, Func<byte[], Task<bool>> fun)
        {
            if (fun == null) throw new ArgumentNullException("fun");
            var section = Configuration.GetSection("RabbitMQ:Consumers:" + queueName);
            bool autoAck = (section != null && !string.IsNullOrEmpty(section["AutoAck"])) ? Convert.ToBoolean(section["AutoAck"]) : AutoAck;
            UInt16 size = (section != null && !string.IsNullOrEmpty(section["Size"])) ? Convert.ToUInt16(section["Size"]) : (UInt16)1;

            for (UInt16 i = 0; i < size; i++)
            {
                MQIModel item = CreateMQIModel(queueName);
                //创建消费者对象
                var consumer = new EventingBasicConsumer(item.Channel);
                //如果列队有配置aotuack则使用队列的，否则使用全局的
                consumer.Received += async (model, ea) =>
                {
                    // byte[] message = ea.Body.ToArray();//接收到的消息
                    // string msg = Encoding.UTF8.GetString(message);
                    bool result = await fun(ea.Body.ToArray());
                    //Console.WriteLine("接收到消息为:" + Encoding.UTF8.GetString(message));
                    if (!autoAck && result)
                    {
                        //返回消息确认
                        item.Channel.BasicAck(ea.DeliveryTag, false);
                    }
                };
                //消费者开启监听
                item.Channel.BasicConsume(queueName, autoAck, consumer);
            }
        }
    }

    public class MQIModel:IDisposable
    {
        public IModel Channel { get; set; }
        public IBasicProperties DurableProperties { get; set; }
        public string Exchange { get; internal set; }
        public string RoutingKey { get; internal set; }
        // public RabbitMQConnectShareManage.ConnectChannelInfo Conn { get; internal set; }

        public virtual void Dispose()
        {
              
        }

        //public RabbitMQPoolConnection Conn { get; internal set; }

        internal void BasicPublish(byte[] body)
        {
            this.Channel.BasicPublish(this.Exchange, this.RoutingKey, this.DurableProperties, body);
        }
    }
}
